package com.idanchuang.support.etcache.pubsub;

import com.idanchuang.support.etcache.Cache;
import com.idanchuang.support.etcache.EtCacheManager;
import com.idanchuang.support.etcache.local.LocalKey;
import com.idanchuang.support.etcache.pubsub.redis.RedisRemoteEventSubscriber;
import com.idanchuang.support.etcache.remote.RemoteKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 缓存变更事件处理器抽象实现
 * @author yjy
 * Created at 2021/12/23 8:35 下午
 */
public class AbstractRemoteEventHandler implements RemoteEventHandler, InitializingBean {

    private static final Logger logger = LoggerFactory.getLogger(RedisRemoteEventSubscriber.class);
    private final ExecutorService master = new ThreadPoolExecutor(1, 1,
            1, TimeUnit.MINUTES, new LinkedBlockingQueue<>(1),
            (r) -> new Thread(r, "et-event-handler-master"));
    private final ExecutorService worker = new ThreadPoolExecutor(3, 3,
            1, TimeUnit.MINUTES, new LinkedBlockingQueue<>(),
            (r -> new Thread(r, "et-event-handler-worker")));

    private final EtCacheManager manager;
    private final SubEventQueue subEventQueue;

    public AbstractRemoteEventHandler(EtCacheManager manager, SubEventQueue subEventQueue) {
        this.manager = manager;
        this.subEventQueue = subEventQueue;
    }

    @Override
    public void start() {
        master.execute(() -> {
            while (true) {
                try {
                    RemoteEvent event = subEventQueue.poll();
                    worker.execute(() -> {
                        try {
                            handle(event);
                        } catch (Exception e) {
                            logger.error("onMessage failed", e);
                        }
                    });
                } catch (Exception e) {
                    logger.error("RemoteEventHandle failed", e);
                }
            }
        });
    }

    @Override
    public void handle(RemoteEvent event) {
        RemoteKey remoteKey = event.getRemoteKey();
        Cache<?, ?> cache = this.manager.getCache(event.getCacheName(), false);
        if (cache == null) {
            logger.warn("handle remote event cancel, cause EtCache is not exist, cacheName: {}", event.getCacheName());
            return;
        }
        Object key = cache.getKeyCodec().decode(remoteKey.getKbs());
        LocalKey localKey = new LocalKey<>(remoteKey.getPressure(), key);
        cache.refreshLocal(localKey);
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        // 启动
        start();
    }

    public EtCacheManager getManager() {
        return manager;
    }

    public SubEventQueue getSubEventQueue() {
        return subEventQueue;
    }
}
